package com.lesent.kafka.service.impl;

import com.lesent.kafka.service.IKafkaService;
import com.lesent.model.entity.Student;
import jdk.nashorn.internal.ir.annotations.Ignore;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author shengtao
 * @Description:
 * @date 2018/08/08 14:01
 */
@Service
public class KafkaServiceImpl implements IKafkaService {
    private final Logger LOGGER = LoggerFactory.getLogger(KafkaServiceImpl.class);
    /**
     * 引入kafka模板实例
     */
    @Autowired
    private KafkaTemplate<String,Student> kafkaTemplate;


    @Override
    public List<String> sendMsg(String s, Map<String,Student> map) {
        final int count = 0;
        map.keySet().stream()
                .peek(key->{
                    ProducerRecord<String,Student> record = new ProducerRecord<String, Student>(
                            "topic",key,map.get(key)
                    );
                    ListenableFuture<SendResult<String, Student>> future = kafkaTemplate.send(record);
                    future.addCallback(new ListenableFutureCallback<SendResult<String, Student>>() {
                        @Override
                        public void onFailure(Throwable throwable) {
                            System.out.println("失败重试");

                        }

                        @Override
                        public void onSuccess(@Nullable SendResult<String, Student> stringStudentSendResult) {
                            RecordMetadata recordMetadata = stringStudentSendResult.getRecordMetadata();
                            System.out.println("成功"+recordMetadata.topic()+"  offset:"+recordMetadata.offset());
                        }
                    });
                }).collect(Collectors.toList());
        return null;
    }


}
